package testing

import (
	"context"
	"encoding/json"
	"fmt"
	"sync"
	"time"

	"github.com/google/uuid"
	"github.com/louloulin/dataflare/pkg/workflow/connector"
)

// MockSourceConnector represents a mock source connector
type MockSourceConnector struct {
	// ID is the connector ID
	ID string
	// data is the mock data
	data []interface{}
	// records is the mock records
	records []*connector.Record
	// position is the current position in the records
	position int
	// metrics are the connector metrics
	metrics map[string]interface{}
	// mu is the mutex
	mu sync.RWMutex
	// started indicates if the connector is started
	started bool
}

// NewMockSourceConnector creates a new mock source connector
func NewMockSourceConnector(data []interface{}) *MockSourceConnector {
	return &MockSourceConnector{
		ID:      uuid.New().String(),
		data:    data,
		metrics: make(map[string]interface{}),
	}
}

// Initialize initializes the connector
func (c *MockSourceConnector) Initialize(config map[string]interface{}) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Initialize metrics
	c.metrics["records_total"] = len(c.data)
	c.metrics["records_read"] = 0
	c.metrics["start_time"] = time.Now().Unix()

	return nil
}

// Start starts the connector
func (c *MockSourceConnector) Start(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if already started
	if c.started {
		return fmt.Errorf("connector already started")
	}

	// Convert data to records
	c.records = make([]*connector.Record, 0, len(c.data))
	for i, item := range c.data {
		// Convert item to JSON
		jsonBytes, err := json.Marshal(item)
		if err != nil {
			return fmt.Errorf("failed to marshal item: %w", err)
		}

		// Create record
		record := connector.NewRecord([]byte(fmt.Sprintf("%d", i)), jsonBytes)
		record.Timestamp = time.Now()
		c.records = append(c.records, record)
	}

	c.started = true

	return nil
}

// Stop stops the connector
func (c *MockSourceConnector) Stop(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if already stopped
	if !c.started {
		return fmt.Errorf("connector not started")
	}

	c.started = false

	return nil
}

// Read reads records from the source
func (c *MockSourceConnector) Read(ctx context.Context) (*connector.RecordBatch, error) {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if started
	if !c.started {
		return nil, fmt.Errorf("connector not started")
	}

	// Check if all records have been read
	if c.position >= len(c.records) {
		return nil, nil
	}

	// Get batch size
	batchSize := 10
	if size, ok := c.metrics["batch_size"].(int); ok {
		batchSize = size
	}

	// Read records
	end := c.position + batchSize
	if end > len(c.records) {
		end = len(c.records)
	}
	batch := c.records[c.position:end]
	c.position = end

	// Update metrics
	c.metrics["records_read"] = c.position
	c.metrics["last_read_time"] = time.Now().Unix()

	return connector.NewRecordBatch(batch), nil
}

// Commit commits the offset
func (c *MockSourceConnector) Commit(ctx context.Context, offset interface{}) error {
	// Mock source does not support commit
	return nil
}

// GetMetrics gets connector metrics
func (c *MockSourceConnector) GetMetrics() map[string]interface{} {
	c.mu.RLock()
	defer c.mu.RUnlock()

	// Copy metrics
	metrics := make(map[string]interface{})
	for k, v := range c.metrics {
		metrics[k] = v
	}

	return metrics
}

// MockSinkConnector represents a mock sink connector
type MockSinkConnector struct {
	// ID is the connector ID
	ID string
	// records are the received records
	records []*connector.Record
	// metrics are the connector metrics
	metrics map[string]interface{}
	// mu is the mutex
	mu sync.RWMutex
	// started indicates if the connector is started
	started bool
}

// NewMockSinkConnector creates a new mock sink connector
func NewMockSinkConnector() *MockSinkConnector {
	return &MockSinkConnector{
		ID:      uuid.New().String(),
		records: make([]*connector.Record, 0),
		metrics: make(map[string]interface{}),
	}
}

// Initialize initializes the connector
func (c *MockSinkConnector) Initialize(config map[string]interface{}) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Initialize metrics
	c.metrics["records_total"] = 0
	c.metrics["start_time"] = time.Now().Unix()

	return nil
}

// Start starts the connector
func (c *MockSinkConnector) Start(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if already started
	if c.started {
		return fmt.Errorf("connector already started")
	}

	c.started = true

	return nil
}

// Stop stops the connector
func (c *MockSinkConnector) Stop(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if already stopped
	if !c.started {
		return fmt.Errorf("connector not started")
	}

	c.started = false

	return nil
}

// Write writes records to the sink
func (c *MockSinkConnector) Write(ctx context.Context, batch *connector.RecordBatch) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if started
	if !c.started {
		return fmt.Errorf("connector not started")
	}

	// Check if batch is valid
	if batch == nil || len(batch.Records) == 0 {
		return nil
	}

	// Store records
	c.records = append(c.records, batch.Records...)

	// Update metrics
	c.metrics["records_total"] = len(c.records)
	c.metrics["last_write_time"] = time.Now().Unix()

	return nil
}

// Flush flushes any buffered records
func (c *MockSinkConnector) Flush(ctx context.Context) error {
	// Mock sink does not need to flush
	return nil
}

// GetMetrics gets connector metrics
func (c *MockSinkConnector) GetMetrics() map[string]interface{} {
	c.mu.RLock()
	defer c.mu.RUnlock()

	// Copy metrics
	metrics := make(map[string]interface{})
	for k, v := range c.metrics {
		metrics[k] = v
	}

	return metrics
}

// GetRecords gets the received records
func (c *MockSinkConnector) GetRecords() []*connector.Record {
	c.mu.RLock()
	defer c.mu.RUnlock()

	// Copy records
	records := make([]*connector.Record, len(c.records))
	copy(records, c.records)

	return records
}

// MockProcessor represents a mock processor
type MockProcessor struct {
	// ID is the processor ID
	ID string
	// processor is the processor function
	processor func(record *connector.Record) *connector.Record
}

// NewMockProcessor creates a new mock processor
func NewMockProcessor(processor func(record *connector.Record) *connector.Record) *MockProcessor {
	return &MockProcessor{
		ID:        uuid.New().String(),
		processor: processor,
	}
}

// Process processes a record
func (p *MockProcessor) Process(record *connector.Record) *connector.Record {
	return p.processor(record)
}
